package com.bw.fun;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.java.tuple.Tuple9;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Collections;

public class Mn6AsyncIOFunc3 extends RichAsyncFunction<String,Tuple9<String,String,Double,Integer,Integer,Double,Long,Double,Double>> {
    Connection conn = null;
    PreparedStatement ps1 = null;
    PreparedStatement ps2 = null;
    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        conn = DriverManager.getConnection("jdbc:phoenix:hadoop102:2181");
        ps1 = conn.prepareStatement("select * from dim_base_organ_yk9 where id=?" );//机构表
        ps2 = conn.prepareStatement("select * from dim_base_region_info_yk9 where id=?" ); //行政区域表
        conn.setAutoCommit(true);
    }
    @Override
    public void close() throws Exception {
        ps1.close();
        ps2.close();
        conn.close();
    }

    @Override
    public void asyncInvoke(String v , ResultFuture<Tuple9<String,String,Double,Integer,Integer,Double,Long,Double,Double>> resultFuture) throws Exception {

        JSONObject data = JSON.parseObject(v).getJSONObject("data");
        // j机构ID
        String org_id = data.getString("start_org_id");
        String cityName = null;
        ps1.setString(1,org_id);
        ResultSet rs1 = ps1.executeQuery();
        if(rs1.next()){
            String org_name = JSON.parseObject(rs1.getString("line")).getString("org_name");
            String region_id = JSON.parseObject(rs1.getString("line")).getString("region_id");
            //查询城市
            ps2.setString(1,region_id.substring(0,4)+"00");
            ResultSet rs2 = ps2.executeQuery();
            if(rs2.next()){
                cityName = JSON.parseObject(rs2.getString("line")).getString("name");
            }
        }
        // 当前日期
        String day = data.getString("create_time").substring(0, 10);
        Double trans_dists = data.getDouble("actual_distance");
         Long  trans_ts = data.getLong("actual_end_time") - data.getLong("actual_start_time");

        resultFuture.complete(Collections.singleton(new Tuple9<>(day,cityName,0.0,0,1,trans_dists,trans_ts,0.0,0.0)));
    }
}
